package com.zhouyu.api;

import com.zhouyu.config.listener.ItemReadEreorlistenler;
import com.zhouyu.config.listener.ItemSaveDbErrorListener;
import com.zhouyu.config.processor.AlipayDbProcessor;
import com.zhouyu.config.processor.AlipayItemProcessor;
import com.zhouyu.dto.AlipayTranDto;
import com.zhouyu.dto.CommonPayOutDto;
import com.zhouyu.entity.CommonPayEntity;
import com.zhouyu.service.AlipayFileReader;
import com.zhouyu.service.AlipayFileWriter;
import com.zhouyu.service.AlipayJpaItemWriter;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import lombok.extern.log4j.Log4j2;

/**
 * @Description:JOb工厂类
 */
@Component
@Log4j2
public class JobCenter {

  @Autowired
  public JobBuilderFactory jobBuilderFactory;

  @Autowired
  public StepBuilderFactory stepBuilderFactory;

  @Autowired
  private AlipayFileReader alipayFileReader;

  @Autowired
  private AlipayJpaItemWriter alipayJpaItemWriter;

  @Autowired
  private AlipayItemProcessor alipayItemProcessor;

  @Autowired
  AlipayDbProcessor alipayDbProcessor;

  @Autowired
  private AlipayFileWriter alipayFileWriter;

  @Autowired
  ItemReadEreorlistenler itemReadListener;

  @Autowired
  ItemSaveDbErrorListener itemSaveDbErrorListener;

  @Bean
  public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setMaxPoolSize(4);
    taskExecutor.afterPropertiesSet();
    return taskExecutor;
  }

  /**
   * 简单的导入
   */
  public Job importSimpleJob(String job) {
    return jobBuilderFactory.get(job)
            .incrementer(new RunIdIncrementer())
            .flow(step1())
            .end()
            .build();
  }

  private Step step1() {
    return stepBuilderFactory.get("step1")
            .<AlipayTranDto, CommonPayOutDto>chunk(10)
//            .reader(alipayFileReader.readFilesByPath("./data/alipay/208012345_20141030.csv"))
            .reader(alipayFileReader.readFiles())
            .processor(alipayItemProcessor)
            .writer(alipayFileWriter.writer())
            .build();
  }

  /**
   * JobFlowBuilder jobBuilderFactory:提供JobBuilder，而且提供JobBuilder也提供了两种 1.SimpleJobBuilder:将执行有序步骤或并行步骤
   * 2.JobFlowBuilder:执行流式步骤
   */
  public Job importThreadJob(String job) {
    return jobBuilderFactory.get(job)
            .start(step2())
            .start(step3())
            .build();
  }

  private Step step2() {
    log.info("****************************开始执行step2");
    return stepBuilderFactory.get("thredstep1")
            .<AlipayTranDto, CommonPayOutDto>chunk(10)
            .reader(alipayFileReader.readFilesByPath("./data/alipay/208012345_20141030.csv"))
            .processor(alipayItemProcessor)
            .writer(alipayFileWriter.writer())
            //在这里设置线程池
            .taskExecutor(taskExecutor())
            .build();
  }

  private Step step3() {
    log.info("****************************开始执行step3");
    return stepBuilderFactory.get("thredstep2")
            .<AlipayTranDto, CommonPayOutDto>chunk(10)
            .reader(alipayFileReader.readFilesByPath("./data/alipay/208054321_20141030.csv"))
            .processor(alipayItemProcessor)
            .writer(alipayFileWriter.writer())
            //在这里设置线程池
            .taskExecutor(taskExecutor())
            .build();
  }


  /**
   * 定义job2，导出到数据库
   */
  public Job importDbJob(String job) {
    return jobBuilderFactory.get(job)
            .incrementer(new RunIdIncrementer())
            .flow(step4())
            .end()
            .build();
  }

  private Step step4() {
    return stepBuilderFactory.get("step4")
            .<AlipayTranDto, CommonPayEntity>chunk(30)
            .reader(alipayFileReader.readFiles())
            .faultTolerant()
            //读取数据监听
            .listener(itemReadListener)
            //加工转化器dto->pojo
            .processor(alipayDbProcessor)
            //使用默认的JpaItemWriter,将pojo插入数据库
            .writer(alipayJpaItemWriter.repositoryItemWriter())
            .faultTolerant()
            .skipLimit(20)
            //在这里应该是自定义异常，如果遇到自定义异常，就要跳过不影响整个step运行
            .skip(Exception.class)
            .retryLimit(2)
            .retry(RuntimeException.class)
            //设置跳过的监听，对于跳过的我们要监听记录下来
            .listener(itemSaveDbErrorListener)
            .build();
  }

}
